package flinkdemo.kafak;


import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.setting.Setting;
import com.arcsoft.face.FaceFeature;
import com.arcsoft.face.FaceInfo;
import com.arcsoft.face.FaceSimilar;
import com.arcsoft.face.enums.DetectMode;
import com.arcsoft.face.enums.DetectOrient;
import com.arcsoft.face.enums.ImageFormat;
import com.google.common.collect.Lists;
import flinkdemo.kafak.conf.EngineConfigurationSer;
import flinkdemo.kafak.entity.FindFaceInfo;
import flinkdemo.kafak.entity.FindMsg;
import flinkdemo.kafak.entity.ProcessInfo;
import flinkdemo.kafak.entity.VideoMsg;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;


/**
 * 初始化人脸引擎，加载人脸库数据，这个也不合理。一次加载后人脸库数据就无法改变了。
 * 后期改成提取人脸向量到milvus向量引擎中识别，返回结果
 * @author Administrator
 */
//public class ImgFlatMap extends RichFlatMapFunction<Tuple2<String, VideoMsg>, Tuple2<String,VideoMsg>> {
public class ImgFlatMap extends RichFlatMapFunction<Tuple2<String, VideoMsg>,FindMsg> {

    public int passRate;
    private String appId = "BGFUM4yT5M9PRGfd9VgUXMzGbK1jZJer4YJzG77HovXR";
    private String sdkKey = "AGURJP2c7PRahrMsS9fPFqW6dFGrgXpdQpELRFEyhtoS";
    private String dllPath = "";


    private Integer detectFaceMaxNum=10;
    private Integer detectFaceScaleVal=16;
    private DetectMode detectMode= DetectMode.ASF_DETECT_MODE_IMAGE;
    private DetectOrient detectFaceOrientPriority= DetectOrient.ASF_OP_0_ONLY;
    private FunctionConfigurationSer functionConfiguration;

    private EngineConfigurationSer engineConfiguration;

    private FaceEngine faceEngine;

    private List<Pair<String,Object>> hashMap;

    ImgFlatMap(String dllPath,String appId, String sdkKey,int passRate) throws Exception {
        this.dllPath = dllPath;
        this.appId = appId;
        this.sdkKey = sdkKey;
        this.passRate = passRate;
        this.hashMap = facesInDb();
    }

    ImgFlatMap() throws Exception {
        this.hashMap = facesInDb();
    }

    public List<Pair<String,Object>> facesInDb() throws Exception {

        List<Pair<String,Object>> hashMap = new ArrayList<>();
        //自定义数据库Setting，更多实用请参阅Hutool-Setting章节
        Setting setting = new Setting("flinkface/db.setting");
        //获取指定配置，第二个参数为分组，用于多数据源，无分组情况下传null
        DataSource ds = DSFactory.create(setting).getDataSource();

        List<Entity> faceFeatures = Db.use(ds).query("select FACE_ID,FACE_FEATURE from user_face_info where checked = '0' ");
        if (CollectionUtils.isNotEmpty(faceFeatures)){
            for (Entity face : faceFeatures){
                hashMap.add(Pair.of(String.valueOf(face.get("FACE_ID")), face.get("FACE_FEATURE")));
            }
        }
        return hashMap;
    }


    @Override
    public void flatMap(Tuple2<String, VideoMsg> value, Collector<FindMsg> out) throws Exception {

        VideoMsg imageInfo = value.f1;
        List<ProcessInfo> infos = extractFacesFeature(imageInfo);
        //筛选出有人脸的
        if (CollectionUtils.isEmpty(infos) || CollectionUtils.isEmpty(hashMap)) {
            return;
        }
        // 一张人脸图片上有 多个人脸，每个人脸 有多个相似脸
        List<List<FindFaceInfo>> finds = new ArrayList<>();
        for(ProcessInfo info : infos){
            List<FindFaceInfo> find = new ArrayList<>();
            for(Pair<String,Object> fc : hashMap ){
                FaceSimilar similar = new FaceSimilar();
                FaceFeature dBFeatrue = new FaceFeature((byte[]) fc.getRight());
                faceEngine.compareFaceFeature(info.getFaceFeature(),dBFeatrue,similar);
                // 获取相似值
                int similarValue = plusHundred(similar.getScore());
                // 相似值大于配置预期，加入到识别到人脸的列表
                if (similarValue > passRate) {
                    FindFaceInfo finfo = new FindFaceInfo();
                    finfo.setFaceId(info.getFace().getFaceId());
                    finfo.setOrient(info.getFace().getOrient());
                    finfo.setRect(info.getFace().getRect());
                    finfo.setSimilarValue(similarValue);
                    find.add(finfo);
                    //log.debug("task线程：" + Thread.currentThread().getName() + "相似度=" + similarValue);
                }
            }
            finds.add(find);
        }
        //清理到人脸识别的数据
        imageInfo.setRgb(null);
        if(CollectionUtils.isEmpty(finds)){
            return;
        }
        FindMsg msg = new FindMsg();
        msg.setCameraId(imageInfo.getCameraId());
        msg.setGroupId(imageInfo.getGroupId());
        msg.setImg(imageInfo.getImg());
        msg.setInfos(finds);
        out.collect(msg);
    }

    private List<ProcessInfo> extractFacesFeature(VideoMsg imageInfo) {
        if(imageInfo == null) {
            return null;
        }
        try{
            // 人脸检测得到人脸列表
            List<FaceInfo> facesInImg = Lists.newArrayList();
            // 图片的长宽和数据
            final byte[] imgByts = imageInfo.getRgb();
            final int width = imageInfo.getWidth();
            final int height = imageInfo.getHeight();
            //int code =
            faceEngine.detectFaces(imgByts, width, height, ImageFormat.CP_PAF_BGR24, facesInImg);

            if (CollectionUtil.isEmpty(facesInImg)) {
                return null;
            }

            List<ProcessInfo> features = Lists.newArrayList();
            for (FaceInfo face : facesInImg) {
                FaceFeature faceFeature = new FaceFeature();
                faceEngine.extractFaceFeature(imgByts, width, height, ImageFormat.CP_PAF_BGR24, face, faceFeature);
                if (faceFeature.getFeatureData() != null){
                    ProcessInfo info = new ProcessInfo();
                    info.setFace(face);
                    info.setFaceFeature(faceFeature);
                    features.add(info);
                }
            }
            return features;
        } catch (Exception e) {
            e.printStackTrace();

        }
        return null;
    }

    /** 乘以100 */
    private int plusHundred(Float value) {
        return new BigDecimal(value).multiply(new BigDecimal(100f)).intValue();
    }



    @Override
    public void open(Configuration parameters) throws Exception {
       /* ClassPathResource resource1 = new ClassPathResource("dll/libarcsoft_face.dll");
        ClassPathResource resource2 = new ClassPathResource("dll/libarcsoft_face_engine.dll");
        ClassPathResource resource3 = new ClassPathResource("dll/libarcsoft_face_engine_jni.dll");

        Path temp1 = Files.createTempFile("libarcsoft_face", "dll");
        Path temp2 = Files.createTempFile("libarcsoft_face_engine", "dll");
        Path temp3 = Files.createTempFile("libarcsoft_face_engine_jni", "dll");
        Files.copy(resource1.getStream(), temp1, StandardCopyOption.REPLACE_EXISTING);
        Files.copy(resource2.getStream(), temp2, StandardCopyOption.REPLACE_EXISTING);
        Files.copy(resource3.getStream(), temp3, StandardCopyOption.REPLACE_EXISTING);*/
        System.out.println("初始化");

        faceEngine = new FaceEngine(dllPath);
        faceEngine.activeOnline(appId,sdkKey);
        engineConfiguration = new  EngineConfigurationSer();

        functionConfiguration = new FunctionConfigurationSer();
        functionConfiguration.setSupportFaceDetect(true);
        functionConfiguration.setSupportFaceRecognition(true);

        engineConfiguration.setFunctionConfiguration(functionConfiguration);
        engineConfiguration.setDetectFaceMaxNum(detectFaceMaxNum);
        engineConfiguration.setDetectFaceScaleVal(detectFaceScaleVal);
        engineConfiguration.setDetectMode(detectMode);
        engineConfiguration.setDetectFaceOrientPriority(detectFaceOrientPriority);

        faceEngine.init(engineConfiguration);
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }


}